package com.smh;

import org.junit.jupiter.api.Test;

import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author shiminghui
 * @date 2025/2/18 20:01
 * @description: TODO
 */
public class _028_阻塞队列_双锁 {


    @Test
    public void test1() throws InterruptedException {
        BlockingQueue<Integer> blockingQueue = new BlockingQueue<>(10);
        for (int i = 0; i < 10; i++) {
            blockingQueue.offer(i);
        }
        new Thread(() -> {
            try {
                blockingQueue.offer(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "t1").start();
        new Thread(() -> {
            try {
                blockingQueue.poll();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "t2").start();

        // 休眠1秒
        TimeUnit.SECONDS.sleep(1);
        for (Integer t : blockingQueue) {
            System.out.println(t);
        }
    }


    @Test
    public void test2() throws InterruptedException {

        BlockingQueue<Integer> queue = new BlockingQueue<>(5);

        // 生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    System.out.println("Producing: " + i);
                    queue.offer(i); // 如果队列已满，这里会阻塞
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    Integer item = queue.poll(); // 如果队列为空，这里会阻塞
                    System.out.println("Consuming: " + item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 启动线程
        producer.start();
        consumer.start();

        try {
            producer.join();
            consumer.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @Test
    public void test3() {

        ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);

        // 生产者线程
        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    System.out.println("Producing: " + i);
                    queue.put(i); // 如果队列已满，这里会阻塞
                }
            } catch (Exception e) {
                Thread.currentThread().interrupt();
            }
        });

        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    Integer item = queue.take(); // 如果队列为空，这里会阻塞
                    System.out.println("Consuming: " + item);
                }
            } catch (Exception e) {
                Thread.currentThread().interrupt();
            }
        });

        // 启动线程
        producer.start();
        consumer.start();

        try {
            producer.join();
            consumer.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private static class BlockingQueue<T> implements com.smh.interfaces.BlockingQueue<T>, Iterable<T> {

        private T[] array;
        private AtomicInteger size = new AtomicInteger();
        private int head;
        private int tail;
        private int capacity;
        private ReentrantLock headLock = new ReentrantLock(); // 锁对象，可重入锁
        private Condition notEmpty = headLock.newCondition(); // 等待不为空
        private ReentrantLock tailLock = new ReentrantLock();
        private Condition notFull = tailLock.newCondition(); // 等待不满


        public BlockingQueue(int capacity) {
            this.capacity = capacity;
            array = (T[]) new Object[capacity];
        }

        public BlockingQueue() {
            this(10);
        }

        @Override
        public void offer(T value) throws InterruptedException {
            tailLock.lockInterruptibly();
            int c;
            try {
                while (isFull()) {
                    notFull.await();
                }
                array[tail++] = value;
                tail = tail % capacity;
                c = size.getAndIncrement(); // 线程安全

                if (c < capacity - 1) { // 级联唤醒 当队列中存在空位时，唤醒等待不为满的线程
                    notFull.signal();
                }
            } finally {
                tailLock.unlock();
            }
            if (c == 0) { // 减少使用锁的次数，只在队列为空时，唤醒等待不为空的线程
                try { // 一个锁的条件变量只能在锁内使用，所以，这里需要再获取锁
                    headLock.lockInterruptibly();
                    notEmpty.signal(); // 唤醒等待不为空的线程
                } finally {
                    headLock.unlock();
                }
            }
        }

        @Override
        public boolean offer(T value, long timeout) throws InterruptedException {
            tailLock.lockInterruptibly();
            int c;
            try {
                long t = TimeUnit.MILLISECONDS.toNanos(timeout);
                while (isFull()) {
                    t = notFull.awaitNanos(t);
                }
                array[tail++] = value;
                tail = tail % capacity;
                c = size.getAndIncrement(); // 线程安全


                if (c < capacity - 1) { // 级联唤醒 当队列中存在空位时，唤醒等待不为满的线程
                    notFull.signal();
                }

            } finally {
                tailLock.unlock();
            }
            if (c == 0) { // 减少使用锁的次数，只在队列为空时，唤醒等待不为空的线程
                try { // 一个锁的条件变量只能在锁内使用，所以，这里需要再获取锁
                    headLock.lockInterruptibly();
                    notEmpty.signal(); // 唤醒等待不为空的线程
                } finally {
                    headLock.unlock();
                }
            }
            return true;
        }

        @Override
        public T poll() throws InterruptedException {
            headLock.lockInterruptibly();
            int c;
            T t;
            try {
                while (isEmpty()) {
                    notEmpty.await();
                }
                t = array[head++];
                array[head - 1] = null;
                head = head % capacity;
                c = size.getAndDecrement(); // 线程安全


                if (c > 1) { // 级联唤醒 当队列中存在多个元素时，唤醒等待不为空的线程
                    notEmpty.signal();
                }

            } finally {
                headLock.unlock();
            }

            if (c == capacity) { // 减少使用锁的次数，只在队列为满时，唤醒等待不满的线程
                try {
                    tailLock.lockInterruptibly();
                    notFull.signal(); // 唤醒等待不满的线程
                } finally {
                    tailLock.unlock();
                }
            }

            return t;
        }

        public boolean isEmpty() {
            return size.get() == 0;
        }

        public boolean isFull() {
            return size.get() == capacity;
        }

        @Override
        public Iterator<T> iterator() {
            return new Iterator<T>() {
                int index = head;
                int count = size.get();

                @Override
                public boolean hasNext() {
                    if (count != 0) {
                        return true;
                    }
                    return false;
                }

                @Override
                public T next() {
                    T value = array[index++];
                    index = index % capacity;
                    count--;
                    return value;
                }
            };
        }
    }

}
